package com.shujia.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

object Demo7OnHIve {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("onhive")
      .config("spark.sql.shuffle.partitions", 1)
      .enableHiveSupport() //在代码中开启hiv元数据支持，后面写代码可以直接使用hive中的表
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      * 通过表名得到一个DF
      *
      */
    val student: DataFrame = spark.table("student")


    /**
      *
      * DSL
      */
    val df: DataFrame = student
      .groupBy($"clazz")
      .agg(count($"id"))


    /**
      * 直接编写sql 使用hive中的表
      *
      */
    /*val df: DataFrame = spark.sql(
      """
        |select clazz,count(1) from student group by clazz
        |
      """.stripMargin)*/


    df.write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save("/data/on_hive")

    /**
      * 将DF 转换成RDD
      *
      */

    val rdd: RDD[Row] = student.rdd

    /**
      * 1、通过列名 获取数据
      *
      */
    val rdd2: RDD[String] = rdd.map(row => {
      //通过列名和类型获取字段
      val clazz: String = row.getAs[String]("clazz")
      clazz
    })

    /**
      * 2、使用模式匹配的方式
      *
      */

    val rdd3: RDD[(String, String, Int, String, String)] = rdd.map {
      case Row(id: String, name: String, age: Int, gender: String, clazz: String) => {
        (id, name, age, gender, clazz)
      }
    }

    rdd3.saveAsTextFile("/data/rdd3")

  }
}
